package org.example.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import lombok.SneakyThrows;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.example.common.PageResponseResult;
import org.example.common.ResponseResult;
import org.example.common.TextModerationPlusDemo;
import org.example.pojo.GoodDto;
import org.example.pojo.Goods;
import org.example.service.GoodsService;
import org.example.mapper.GoodsMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

/**
* @author 贾媛婕
* @description 针对表【goods】的数据库操作Service实现
* @createDate 2024-09-05 10:19:03
*/
@Service
public class GoodsServiceImpl extends ServiceImpl<GoodsMapper, Goods>
    implements GoodsService{
    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;

    @Resource
    private RedisTemplate<String,Object> redisTemplate;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Override
    public ResponseResult addGood(Goods goods) {
        this.save(goods);
        return ResponseResult.okResult("添加成功");
    }

    @Override
    public ResponseResult goodList(GoodDto dto) {
        System.err.println(dto);
        System.err.println(!StringUtils.isBlank(dto.getName()));
        System.err.println("========================");
        if(StringUtils.isBlank(dto.getName())){
            Page page = new Page(dto.getPageNum(),dto.getPageSize());
            this.page(page,null);

            ResponseResult result = new PageResponseResult(dto.getPageNum(),dto.getPageSize(), (int) page.getTotal());
            result.setData(page.getRecords());
            return result;
        }


        ResponseResult result = this.searchKeyWord(dto);
        return result;


    }

    @Override
    public ResponseResult delGood(Integer id) {
        return ResponseResult.okResult(this.removeById(id));
    }

    @Override
    public ResponseResult updateGood(Goods goods) {
        boolean update = this.update(Wrappers.<Goods>lambdaUpdate().set(!StringUtils.isBlank(goods.getName()), Goods::getName, goods.getName())
                .set(!StringUtils.isBlank(goods.getDescrible()), Goods::getDescrible, goods.getDescrible())
                .set(goods.getPrice() != null, Goods::getPrice, goods.getPrice())
                .set(Goods::getCreateTime, new Date())
                .set(goods.getPublishTime() != null, Goods::getPublishTime, goods.getPublishTime())
                .set(goods.getIsUp() != null, Goods::getIsUp, goods.getIsUp())
                .set(goods.getIsShen() != null, Goods::getIsShen, goods.getIsShen())
                .eq(Goods::getId, goods.getId()));
        return update?ResponseResult.okResult(200):ResponseResult.errorResult(500,"no update");
    }

    @Override
    public ResponseResult getOne(Integer id) {
        Goods byId = this.getById(id);
        return ResponseResult.okResult(byId);
    }

    @Override
    @SneakyThrows
    public ResponseResult autoGood(Integer id) {
        Goods byId = this.getById(id);
        String goodString = byId.getName()+byId.getDescrible();

        ResponseResult result = TextModerationPlusDemo.autoText(goodString);
        Integer code = result.getCode();
        if(code == 601){
            byId.setIsShen(1);
            this.updateById(byId);
            //同步zset或者kafka中
            tongbu(byId);

            return ResponseResult.okResult(601,"审核通过");
        } else if (code == 602) {
            byId.setIsShen(2);
            this.updateById(byId);
            return ResponseResult.errorResult(602,"审核拒绝");
        }else {
            byId.setIsShen(3);
            this.updateById(byId);
            return ResponseResult.errorResult(603,"人工审核");
        }
    }

    @Override
    @SneakyThrows
    public ResponseResult upGood(Integer id) {
        Goods byId = this.getById(id);
        Integer isShen = byId.getIsShen();
        if (isShen == 1){
            byId.setIsUp(1);
            this.updateById(byId);
            IndexRequest indexRequest = new IndexRequest("goods")
                    .source(JSON.toJSONString(byId), XContentType.JSON).id(byId.getId()+"");

            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);

            return ResponseResult.okResult(200,"上架成功");
        }

        return ResponseResult.errorResult(500,"未审核，无法上架");
    }

    @Override
    @SneakyThrows
    public ResponseResult downGood(Integer id) {
        Goods byId = this.getById(id);
        byId.setIsUp(0);
        this.updateById(byId);
        DeleteRequest deleteRequest = new DeleteRequest("goods").id(id+"");
        restHighLevelClient.delete(deleteRequest,RequestOptions.DEFAULT);

        return ResponseResult.okResult("下架成功");
    }

    @Override
    @SneakyThrows
    public ResponseResult enInit() {
        List<Goods> list = this.list(Wrappers.<Goods>lambdaQuery().eq(Goods::getIsUp, 1));

        BulkRequest bulkRequest = new BulkRequest("goods");

        for (Goods goods : list) {
            IndexRequest indexRequest = new IndexRequest().id(goods.getId()+"").source(JSON.toJSONString(goods),XContentType.JSON);
            bulkRequest.add(indexRequest);
        }
        restHighLevelClient.bulk(bulkRequest,RequestOptions.DEFAULT);

        return null;
    }

    @Override
    @SneakyThrows
    public ResponseResult searchKeyWord(GoodDto dto) {



        SearchRequest searchRequest = new SearchRequest();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();


        QueryStringQueryBuilder queryBuilder = QueryBuilders.queryStringQuery(dto.getName()).field("name").defaultOperator(Operator.OR);



        searchSourceBuilder.query(queryBuilder);
        searchRequest.source(searchSourceBuilder);
        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        SearchHit[] hits = search.getHits().getHits();
        List<Goods> list = new ArrayList<>();
        for (SearchHit hit : hits) {
            String goodJson = hit.getSourceAsString();
            Goods goods = JSON.parseObject(goodJson, Goods.class);
            list.add(goods);
        }

        PageHelper.startPage(dto.getPageNum(),dto.getPageSize());
        PageInfo<Goods> goodsPageInfo = new PageInfo<>(list);


        return ResponseResult.okResult(goodsPageInfo);

//        //创建查询
//        SearchRequest searchRequest = new SearchRequest();
//        SearchSourceBuilder builder = new SearchSourceBuilder();
//        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
//
//        QueryStringQueryBuilder queryBuilder=QueryBuilders.queryStringQuery(keyword).field("name").defaultOperator(Operator.OR);
//
//        boolQueryBuilder.must(queryBuilder);
//        builder.query(queryBuilder);
//        searchRequest.source(builder);
//        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
//
//        SearchHit[] hits = search.getHits().getHits();
//
//        List list=new ArrayList();
////循环响应结果添加到集合当中
//        for (SearchHit hit : hits) {
//            String sourceAsString = hit.getSourceAsString();
//            Goods goods = JSON.parseObject(sourceAsString, Goods.class);
//
//            list.add(goods);
//        }


//        return ResponseResult.okResult(list);
    }

    public void tongbu(Goods goods){
        long current = System.currentTimeMillis();
        long publishTime = goods.getPublishTime().getTime();
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);

        long future = calendar.getTimeInMillis();

        if(publishTime <= current){
            kafkaTemplate.send("publish", JSON.toJSONString(goods));
            System.err.println("添加kafka，立即消费——————————————————");
        } else if (publishTime<= future) {
            redisTemplate.opsForZSet().add("future",goods,publishTime);
            System.err.println("添加到zset中————————————————");
        }

    }

    @Scheduled(cron = "0 0/1 * * * ?")
    public void toKafka(){
        System.out.println("每分钟同步");
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,1);
        long future = calendar.getTimeInMillis();

        Set<Object> future1 = redisTemplate.opsForZSet().rangeByScore("future", 0, future);
        for (Object o : future1) {
            kafkaTemplate.send("publish",JSON.toJSONString(o));
            redisTemplate.opsForZSet().remove("future",o);

            System.out.println("移除zset，添加kafka——————————————");
        }

    }

    @KafkaListener(topics = "publish")
    public void xiaofei(String msg) throws IOException {
        Goods goods = JSON.parseObject(msg, Goods.class);
        goods.setIsUp(1);
        this.updateById(goods);
        System.out.println("消费kafka数据————————————————");

        IndexRequest indexRequest = new IndexRequest("goods")
                .source(JSON.toJSONString(goods), XContentType.JSON).id(goods.getId()+"");

        restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        System.out.println("——————————————————同步到es中————————————————————");


    }

}




